package com.wudl.flink.core;

import com.wudl.flink.bean.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName : Flink_PageView_WordCount
 * @Description :
 * @Author :wudl
 * @Date: 2021-09-21 11:28
 */

public class Flink_PageView_WordCount_Process {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> readTextFile = env.readTextFile("D:\\ideaWorkSpace\\learning\\Flinklearning\\wudl-flink-12\\input\\UserBehavior.csv");
        SingleOutputStreamOperator<UserBehavior> pv = readTextFile.flatMap(new FlatMapFunction<String, UserBehavior>() {
            @Override
            public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                String[] split = value.split(",");
                // 封装
                UserBehavior userBehavior = new UserBehavior(Long.parseLong(split[0]),
                        Long.parseLong(split[1]),
                        Integer.parseInt(split[2]),
                        split[3],
                        Long.parseLong(split[4]));

                if ("pv".equals(userBehavior.getBehavior())) {
                    out.collect(userBehavior);
                }
            }
        });
        // 分组
        KeyedStream<UserBehavior, String> keyedStream = pv.keyBy(data -> "pv");
        // 计算总和
        SingleOutputStreamOperator<Integer> process = keyedStream.process(new KeyedProcessFunction<String, UserBehavior, Integer>() {
            int count = 0;

            @Override
            public void processElement(UserBehavior userBehavior, Context context, Collector<Integer> collector) throws Exception {
                count++;
                System.out.println(userBehavior.toString());
                collector.collect(count);
            }
        });
        process.print();
        env.execute();

    }
}
